package jupitermouse.site.example.project;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.Random;

/**
 * @author qiqiang.ren@hand-china.com
 */
public class MyKafkaProducer {
    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.81.130:9092");
        properties.setProperty("zookeeper.connect", "192.168.81.130:2181");
        properties.setProperty("key.serializer", StringSerializer.class.getName());
        properties.setProperty("value.serializer", StringSerializer.class.getName());

        KafkaProducer<String,String> producer = new KafkaProducer<>(properties);

        String topic = "flink";
        // 通过死循环一直不停通过Kafka的Broker生产数据
        while (true){
            StringBuilder builder = new StringBuilder();
            builder.append("hand").append("\t")
                    .append("CN").append("\t")
                    .append(getLevels()).append("\t")
                    .append(getTimes()).append("\t")
                    .append(getIps()).append("\t")
                    .append(getDomains()).append("\t")
                    .append(getTraffics()).append("\t");

            System.out.println(builder.toString());
            producer.send(new ProducerRecord<>(topic,builder.toString()));
            Thread.sleep(1000);
        }


    }

    public static int getTraffics(){
        return new Random().nextInt(1000);
    }

    public static String getDomains(){
        String[] domains = new String[]{
              "user001.hand.com",
                "user002.hand.com",
                "user003.hand.com",
                "user004.hand.com",
                "user005.hand.com",
                "user006.hand.com",
                "user007.hand.com",
                "user008.hand.com",
                "user009.hand.com",
                "user010.hand.com",
                "user011.hand.com",
                "user012.hand.com",
                "user013.hand.com",
                "user014.hand.com",
                "user015.hand.com",
                "user016.hand.com",
        };
        return domains[new Random().nextInt(domains.length)];
    }

    public static String getIps(){
        String[] ips = new String[]{
                "223.104.18.110",
                "113.101.75.194",
                "183.225.139.16",
                "59.83.198.84",
                "117.59.39.169",
                "117.59.39.160",
                "117.59.39.161",
                "117.59.39.162",
                "117.59.39.163",
                "117.59.39.164",
                "117.59.39.165",
                "117.59.39.166",
                "117.59.39.167",
                "117.59.39.168"
        };
        return ips[new Random().nextInt(ips.length)];
    }

    public static String getTimes(){
        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
                .format(new Date());
    }

    //生成level数据
    public static String getLevels(){
        String[] levels = new String[]{"M","E"};
        return levels[new Random().nextInt(levels.length)];
    }
}
